package com.shujia.flink.transformaction

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{SplitStream, StreamExecutionEnvironment}


object Demo6SplitSelectOperator {
  def main(args: Array[String]): Unit = {


    //创建flink  程序的入口  相当于sparkContext
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    val ds = env.readTextFile("spark/data/students.txt")

    val studentDS = ds.map(line => {
      val split = line.split(",")
      val id = split(0)
      val name = split(1)
      val age = split(2).toInt
      val gender = split(3)
      val clazz = split(4)
      Student(id, name, age, gender, clazz, 1)
    })

    //将性别为男和女的数据分词两个流
    val splitDS: SplitStream[Student] = studentDS.split(student => {
      if (student.gender.equals("男")) {
        List("男") //返回一个标记集合
      } else {
        List("女")
      }
    })

    //通过标记获取流
    //    splitDS.select("男").print()
    splitDS.select("女").print()


    env.execute("KeyBy")


  }

  case class Student(id: String, name: String, age: Int, gender: String, clazz: String, i: Int)

}
